package com.zk;

import com.zk.kafka.CsvDataProducer;
import com.zk.storm.DataAggregationTopology;

public class Application {

    public static void main(String[] args) throws Exception {
        // 启动Kafka数据生产者（从CSV文件发送数据）
        Thread producerThread = new Thread(() -> {
            try {
                CsvDataProducer.main(new String[0]);
            } catch (Exception e) {
                System.err.println("Kafka数据生产者启动失败:");
                e.printStackTrace();
            }
        });
        producerThread.start();
        System.out.println("Kafka数据生产者已启动");

        // 启动Storm聚合拓扑
        Thread stormThread = new Thread(() -> {
            try {
                DataAggregationTopology.main(new String[0]);
            } catch (Exception e) {
                System.err.println("Storm聚合拓扑启动失败:");
                e.printStackTrace();
            }
        });
        stormThread.start();
        System.out.println("Storm聚合拓扑已启动");

        // 等待生产者和拓扑线程完成
        producerThread.join();
        stormThread.join();
    }
}